/*
 * ProducerConsumerTest.cpp
 *
 *  Created on: Feb 9, 2010
 *      Author: Jamie LaPointe
 */

#include "ProducerConsumerTest.h"
#include "pthreadHelper.h"
#include <iostream>
#include <sstream>
#include <semaphore.h>
#include <cstring>

namespace ProducerConsumer
{
static sem_t exitSem;
}

using namespace ProducerConsumer;

void * ProducerConsumerTest::producerTest(void * arg)
{
   TestData * testData = (TestData *) arg;
   std:: ostringstream stringBuf(std::ios::out);

   for (int i = 0; i < maxIterations; ++i)
   {
      Common::Request req;
      req.isNullTerm = false;
      req.isValid = true;
      req.reqLength = 24;
      req.req = (Char8 *) malloc(req.reqLength * sizeof(Char8));
      strcpy(req.req,"This is my test string.\0");

      stringBuf << "Attempting to produce on Thread# " << testData->threadIndex
            << " iteration# " << i;
      testData->producerConsumerTest->log->writeLine(stringBuf.str());

      // clear the string buffer
      stringBuf.str("");

      testData->producerConsumerTest->producer->run(req);

      stringBuf << "Produced on Thread# " << testData->threadIndex << " iteration# " << i;
      testData->producerConsumerTest->log->writeLine(stringBuf.str());

      // clear the string buffer
      stringBuf.str("");
   }

   sem_post(&exitSem);

   pthread_exit(0);
   return 0;
}

void * ProducerConsumerTest::consumerTest(void * arg)
{
   TestData * testData = (TestData *) arg;
   std::ostringstream stringBuf(std::ios::out);

   for (int i = 0; i < maxIterations; ++i)
   {
      stringBuf << "Attempting to consume on Thread# " << testData->threadIndex
            << " iteration# " << i;
      testData->producerConsumerTest->log->writeLine(stringBuf.str());

      // clear the string buffer
      stringBuf.str("");

      Common::Request consumedReq = testData->producerConsumerTest->consumer->run();

      stringBuf << "Consumed on Thread# " << testData->threadIndex << " iteration# " << i;
      testData->producerConsumerTest->log->writeLine(stringBuf.str());

      // clear the string buffer
      stringBuf.str("");
   }

   sem_post(&exitSem);

   pthread_exit(0);
   return 0;
}

ProducerConsumerTest::ProducerConsumerTest()
{
   sharedData = new Common::ProducerConsumerShared();
   producer = new Producer::Producer(sharedData);
   consumer = new Consumer::Consumer(sharedData);
   log = new Logger::Log();
}

ProducerConsumerTest::~ProducerConsumerTest()
{
   delete (producer);
   producer = 0;

   delete (consumer);
   consumer = 0;

   delete (log);
   log = 0;

   delete (sharedData);
   sharedData = 0;
}

void ProducerConsumerTest::Run()
{
   pthread_t consumerThreads[numConsumerThreads];
   pthread_t producerThreads[numProducerThreads];
   pthread_attr_t threadAttr;
   TestData * consumerTestData[numConsumerThreads];
   TestData * producerTestData[numProducerThreads];

   //
   // init request list
   //
   sharedData->requestList.reserve(
         (std::vector<Common::Request>::size_type) maxSizeList);

   //
   // setup thread attribute
   //
   pthread_attr_init(&threadAttr);
   pthread_attr_setdetachstate(&threadAttr, PTHREAD_CREATE_DETACHED);

   //
   // Initialize the shared exit semaphore - used to wait for worker threads to finish
   // before allowing the main thread (and process) to exit.
   //
   sem_init(&exitSem, 0, 0);

   //
   // Setup print mutex
   //
   //pthread_mutex_init(&printMutex_);

   //
   // Create and execute the consumer threads
   //
   for (int i = 0; i < numConsumerThreads; ++i)
   {
      consumerTestData[i] = new TestData();
      consumerTestData[i]->producerConsumerTest = this;
      consumerTestData[i]->threadIndex = i;

      pthread_create(&consumerThreads[i], &threadAttr, consumerTest,
            (void *) consumerTestData[i]);
   }

   //
   // Create and execute the producer threads
   //
   for (int i = 0; i < numProducerThreads; ++i)
   {
      producerTestData[i] = new TestData();
      producerTestData[i]->producerConsumerTest = this;
      producerTestData[i]->threadIndex = i;

      pthread_create(&producerThreads[i], &threadAttr, producerTest,
            (void *) producerTestData[i]);
   }

   //
   // Cleanup after yourself
   //
   pthread_attr_destroy(&threadAttr);
   //pthread_mutex_destroy(&printMutex);

   // wait for all worker threads to exit before allowing the main thread to exit.
   threadSingleBarrier(&exitSem, numConsumerThreads + numProducerThreads);

   for (int i = 0; i< numConsumerThreads; ++i)
   {
      delete(consumerTestData[i]);
   }
   for (int i = 0; i < numProducerThreads; ++i)
   {
      delete(producerTestData[i]);
   }

   sem_destroy(&exitSem);
}
